package com.company.practice;


import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class Test2 {


    /**
     * 当我们需要批量任务处理,但是并不关心任务完成的先后顺序,我们异步的提交任务,等待有任务执行完成之后然后对该完成结果处理,
     * 如此循环直到该批量任务完成. 我们遵循异步处理完成后的操作原则时,谁先完成收割谁.   Future
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Test
    public void fun1() throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(5);
        final List<String> dList = Arrays.asList("aa", "bb", "cc", "dd", "ee");
        List<Future<Object>> fList = new ArrayList<>();
        for (int i = 0; i < dList.size(); i++) {
            final int tmp = i;
            Future<Object> future = pool.submit(new Callable<Object>() {
                @Override
                public Object call() throws Exception {
                    if (tmp == 2) {
                        Thread.sleep(3000);
                    }
                    Thread.sleep(1000);
                    return "线程" + Thread.currentThread().getName() + "处理数据元素list(" + +tmp + ") = " + dList.get(tmp);
                }
            });
            fList.add(future);
        }
        System.out.println("聚合完成");
        for (int i = 0; i < fList.size(); i++) {
            System.out.println(fList.get(i).get());
        }


    }


    /**
     * 当我们需要批量任务处理,但是并不关心任务完成的先后顺序,我们异步的提交任务,等待有任务执行完成之后然后对该完成结果处理,
     * 如此循环直到该批量任务完成. 我们遵循异步处理完成后的操作原则时,谁先完成收割谁.   CompleteService take()
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Test
    public void fun2() throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(5);
        CompletionService<Object> cs = new ExecutorCompletionService<>(pool);
        final List<String> dList = Arrays.asList("aa", "bb", "cc", "dd", "ee");
        List<Future<Object>> fList = new ArrayList<>();
        for (int i = 0; i < dList.size(); i++) {
            final int tmp = i;
            cs.submit(new Callable<Object>() {
                @Override
                public Object call() throws Exception {
                    if (tmp == 2) {
                        Thread.sleep(3000);
                    }
                    Thread.sleep(1000);
                    return "线程" + Thread.currentThread().getName() + "处理数据元素list(" + +tmp + ") = " + dList.get(tmp);
                }
            });
        }
        System.out.println("聚合完成");
//        for (int i = 0; i < dList.size(); i++) {
//            System.out.println(cs.take().get());
//        }

        while (true) {
            System.out.println(cs.take().get());
        }
    }


    /**
     * 当我们需要批量任务处理,但是并不关心任务完成的先后顺序,我们异步的提交任务,等待有任务执行完成之后然后对该完成结果处理,
     * 如此循环直到该批量任务完成. 我们遵循异步处理完成后的操作原则时,谁先完成收割谁.   CompleteService poll()
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Test
    public void fun3() throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(5);
        CompletionService<Object> cs = new ExecutorCompletionService<Object>(pool);
        final List<String> dList = Arrays.asList("aa", "bb", "cc", "dd", "ee");
        for (int i = 0; i < dList.size(); i++) {
            final int tmp = i;
            cs.submit(new Callable<Object>() {
                @Override
                public Object call() throws Exception {
                    if (tmp == 2) {
                        Thread.sleep(3000);
                    }
                    Thread.sleep(1000);
                    return "线程" + Thread.currentThread().getName() + "处理数据元素list(" + +tmp + ") = " + dList.get(tmp);
                }
            });
        }
        System.out.println("聚合完成");
        AtomicInteger index = new AtomicInteger(0);
        while (index.get() < dList.size()) {
            Future<Object> f = cs.poll();
            if (f == null) {
                System.out.println("没发现有完成的任务");
            } else {
                //只有获取到非null值，index才+1
                System.out.println(f.get());
                index.incrementAndGet();
            }
            Thread.sleep(500);
        }


    }

}
